package com.example.minisqlite;

import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
 * 引擎：基于 StorageEngine 执行 CREATE/INSERT/SELECT/UPDATE/DELETE，并维护二级索引。
 */
public class Engine implements AutoCloseable {

    private final StorageEngine storage;
    private final WalManager wal;
    private Catalog catalog;
    private final Map<Integer, List<SecondaryIndex>> secondaryIndexes = new HashMap<>();

    public Engine(StorageEngine storage, WalManager wal) throws SQLException {
        this.storage = storage;
        this.wal = wal;
        try {
            byte[] meta = storage.loadCatalog();
            if (meta == null) {
                catalog = new Catalog();
                persistCatalog();
            } else {
                catalog = Catalog.deserialize(meta);
            }
            rebuildIndexes();
        } catch (IOException | ClassNotFoundException e) {
            throw new SQLException("Failed to load catalog", e);
        }
    }

    private void persistCatalog() throws SQLException {
        try {
            storage.saveCatalog(catalog.serialize());
        } catch (IOException e) {
            throw new SQLException("Failed to persist catalog", e);
        }
    }

    public synchronized ResultSet createTable(String name, List<ColumnDefinition> columns) throws SQLException {
        validateColumnNames(columns);
        catalog.createTable(name, columns);
        persistCatalog();
        rebuildIndexes();
        return ResultSet.updateCount(0);
    }

    public synchronized ResultSet createIndex(String indexName, String tableName, String columnName) throws SQLException {
        IndexDefinition definition = catalog.createIndex(indexName, tableName, columnName);
        persistCatalog();
        TableDefinition table = catalog.getTable(tableName);
        SecondaryIndex index = buildIndex(table, definition);
        secondaryIndexes.computeIfAbsent(table.getTableId(), id -> new ArrayList<>()).add(index);
        return ResultSet.updateCount(0);
    }

    private void validateColumnNames(List<ColumnDefinition> columns) throws SQLException {
        Set<String> names = new HashSet<>();
        for (ColumnDefinition column : columns) {
            String lower = column.getName().toLowerCase();
            if (!names.add(lower)) {
                throw new SQLException("Duplicate column name: " + column.getName());
            }
        }
    }

    public synchronized ResultSet insert(String tableName, List<String> columnNames, List<Object> values) throws SQLException {
        TableDefinition table = catalog.getTable(tableName);
        if (columnNames.size() != values.size()) {
            throw new SQLException("Column count does not match value count");
        }
        Object[] rowValues = new Object[table.getColumns().size()];
        for (int i = 0; i < columnNames.size(); i++) {
            int idx = table.findColumnIndex(columnNames.get(i));
            if (idx < 0) {
                throw new SQLException("Unknown column: " + columnNames.get(i));
            }
            rowValues[idx] = convertValue(table.getColumns().get(idx).getType(), values.get(i));
        }
        for (int i = 0; i < rowValues.length; i++) {
            if (rowValues[i] == null) {
                rowValues[i] = defaultValue(table.getColumns().get(i).getType());
            }
        }
        long rowId = assignRowId(table, rowValues);
        byte[] payload = RowCodec.encode(table.getColumns(), rowValues);
        try {
            wal.begin();
            wal.logPut(table.getTableId(), rowId, payload);
            wal.commit();
            storage.putRow(table.getTableId(), rowId, payload);
            wal.checkpoint();
            addToIndexes(table, rowId, rowValues);
        } catch (IOException e) {
            throw new SQLException("Failed to persist row", e);
        }
        return ResultSet.updateCount(1);
    }

    private long assignRowId(TableDefinition table, Object[] rowValues) throws SQLException {
        if (table.hasPrimaryKey()) {
            int idx = table.getPrimaryKeyIndex();
            Object value = rowValues[idx];
            if (value == null) {
                throw new SQLException("PRIMARY KEY column cannot be null");
            }
            long rowId = ((Number) value).longValue();
            table.ensureRowId(rowId);
            return rowId;
        }
        return table.nextRowId();
    }

    private Object convertValue(ColumnType type, Object value) throws SQLException {
        if (value == null) {
            return null;
        }
        switch (type) {
            case INTEGER:
                if (value instanceof Number) {
                    return ((Number) value).longValue();
                }
                if (value instanceof String) {
                    try {
                        return Long.parseLong((String) value);
                    } catch (NumberFormatException e) {
                        throw new SQLException("Invalid INTEGER value: " + value);
                    }
                }
                break;
            case REAL:
                if (value instanceof Number) {
                    return ((Number) value).doubleValue();
                }
                if (value instanceof String) {
                    try {
                        return Double.parseDouble((String) value);
                    } catch (NumberFormatException e) {
                        throw new SQLException("Invalid REAL value: " + value);
                    }
                }
                break;
            case TEXT:
                return value.toString();
            default:
                break;
        }
        throw new SQLException("Cannot convert value " + value + " to type " + type);
    }

    private Object defaultValue(ColumnType type) {
        return null;
    }

    public synchronized ResultSet select(String tableName, List<String> projection, Filter filter) throws SQLException {
        TableDefinition table = catalog.getTable(tableName);
        List<String> columns;
        if (projection == null || projection.isEmpty()) {
            columns = getAllColumnNames(table);
        } else {
            columns = projection;
        }
        List<Row> rows = fetchRows(table, filter);
        List<Object[]> projected = new ArrayList<>();
        for (Row row : rows) {
            Object[] projectedValues = new Object[columns.size()];
            for (int i = 0; i < columns.size(); i++) {
                int idx = table.findColumnIndex(columns.get(i));
                if (idx < 0) {
                    throw new SQLException("Unknown column in projection: " + columns.get(i));
                }
                projectedValues[i] = row.getValues()[idx];
            }
            projected.add(projectedValues);
        }
        return ResultSet.rows(columns, projected);
    }

    public synchronized ResultSet update(String tableName, List<String> columnNames, List<Object> values, Filter filter) throws SQLException {
        TableDefinition table = catalog.getTable(tableName);
        if (columnNames.isEmpty()) {
            throw new SQLException("UPDATE requires at least one column");
        }
        if (columnNames.size() != values.size()) {
            throw new SQLException("Column count does not match value count");
        }
        int[] columnIndexes = new int[columnNames.size()];
        Object[] convertedValues = new Object[columnNames.size()];
        for (int i = 0; i < columnNames.size(); i++) {
            int idx = table.findColumnIndex(columnNames.get(i));
            if (idx < 0) {
                throw new SQLException("Unknown column: " + columnNames.get(i));
            }
            columnIndexes[i] = idx;
            convertedValues[i] = convertValue(table.getColumns().get(idx).getType(), values.get(i));
        }
        List<Row> rows = fetchRows(table, null);
        List<RowMutation> mutations = new ArrayList<>();
        for (Row row : rows) {
            if (filter == null || filter.matches(row)) {
                Object[] newValues = row.getValues().clone();
                for (int i = 0; i < columnIndexes.length; i++) {
                    newValues[columnIndexes[i]] = convertedValues[i];
                }
                byte[] payload = RowCodec.encode(table.getColumns(), newValues);
                mutations.add(new RowMutation(row.getRowId(), row.getValues(), newValues, payload));
            }
        }
        if (mutations.isEmpty()) {
            return ResultSet.updateCount(0);
        }
        try {
            wal.begin();
            for (RowMutation mutation : mutations) {
                wal.logPut(table.getTableId(), mutation.rowId(), mutation.payload());
            }
            wal.commit();
            for (RowMutation mutation : mutations) {
                storage.putRow(table.getTableId(), mutation.rowId(), mutation.payload());
                updateIndexesOnUpdate(table, mutation.rowId(), mutation.oldValues(), mutation.newValues());
            }
            wal.checkpoint();
        } catch (IOException e) {
            throw new SQLException("Failed to persist updated rows", e);
        }
        return ResultSet.updateCount(mutations.size());
    }

    public synchronized ResultSet delete(String tableName, Filter filter) throws SQLException {
        TableDefinition table = catalog.getTable(tableName);
        List<Row> rows = fetchRows(table, null);
        List<Row> deletes = new ArrayList<>();
        for (Row row : rows) {
            if (filter == null || filter.matches(row)) {
                deletes.add(row);
            }
        }
        if (deletes.isEmpty()) {
            return ResultSet.updateCount(0);
        }
        try {
            wal.begin();
            for (Row row : deletes) {
                wal.logDelete(table.getTableId(), row.getRowId());
            }
            wal.commit();
            for (Row row : deletes) {
                storage.deleteRow(table.getTableId(), row.getRowId());
                removeFromIndexes(table, row.getRowId(), row.getValues());
            }
            wal.checkpoint();
        } catch (IOException e) {
            throw new SQLException("Failed to delete rows", e);
        }
        return ResultSet.updateCount(deletes.size());
    }

    private List<Row> fetchRows(TableDefinition table, Filter filter) throws SQLException {
        if (filter != null) {
            SecondaryIndex index = findIndex(table.getTableId(), filter.getColumn());
            if (index != null) {
                List<Long> rowIds = index.lookup(filter.getExpected());
                List<Row> result = new ArrayList<>(rowIds.size());
                for (Long rowId : rowIds) {
                    StorageEngine.RowRecord record = storage.getRowRecord(table.getTableId(), rowId);
                    if (record == null) {
                        continue;
                    }
                    Object[] values = RowCodec.decode(table.getColumns(), record.payload());
                    Row row = new Row(table, rowId, values);
                    if (filter.matches(row)) {
                        result.add(row);
                    }
                }
                return result;
            }
        }
        List<StorageEngine.RowRecord> records = storage.scanTable(table.getTableId());
        List<Row> result = new ArrayList<>(records.size());
        for (StorageEngine.RowRecord record : records) {
            Object[] values = RowCodec.decode(table.getColumns(), record.payload());
            Row row = new Row(table, record.rowId(), values);
            if (filter == null || filter.matches(row)) {
                result.add(row);
            }
        }
        return result;
    }

    private List<String> getAllColumnNames(TableDefinition table) {
        List<String> names = new ArrayList<>();
        for (ColumnDefinition column : table.getColumns()) {
            names.add(column.getName());
        }
        return names;
    }

    private void rebuildIndexes() throws SQLException {
        secondaryIndexes.clear();
        for (TableDefinition table : catalog.getTables()) {
            List<IndexDefinition> defs = catalog.getIndexesForTable(table.getTableId());
            if (defs.isEmpty()) {
                continue;
            }
            List<SecondaryIndex> indexes = new ArrayList<>();
            for (IndexDefinition def : defs) {
                indexes.add(buildIndex(table, def));
            }
            secondaryIndexes.put(table.getTableId(), indexes);
        }
    }

    private SecondaryIndex buildIndex(TableDefinition table, IndexDefinition definition) throws SQLException {
        ColumnDefinition column = table.getColumns().get(definition.getColumnIndex());
        SecondaryIndex index = new SecondaryIndex(definition, column);
        List<StorageEngine.RowRecord> records = storage.scanTable(table.getTableId());
        for (StorageEngine.RowRecord record : records) {
            Object[] values = RowCodec.decode(table.getColumns(), record.payload());
            index.add(values[definition.getColumnIndex()], record.rowId());
        }
        return index;
    }

    private void addToIndexes(TableDefinition table, long rowId, Object[] values) {
        List<SecondaryIndex> indexes = secondaryIndexes.get(table.getTableId());
        if (indexes == null) {
            return;
        }
        for (SecondaryIndex index : indexes) {
            index.add(values[index.getColumnIndex()], rowId);
        }
    }

    private void removeFromIndexes(TableDefinition table, long rowId, Object[] values) {
        List<SecondaryIndex> indexes = secondaryIndexes.get(table.getTableId());
        if (indexes == null) {
            return;
        }
        for (SecondaryIndex index : indexes) {
            index.remove(values[index.getColumnIndex()], rowId);
        }
    }

    private void updateIndexesOnUpdate(TableDefinition table, long rowId, Object[] oldValues, Object[] newValues) {
        List<SecondaryIndex> indexes = secondaryIndexes.get(table.getTableId());
        if (indexes == null) {
            return;
        }
        for (SecondaryIndex index : indexes) {
            int columnIndex = index.getColumnIndex();
            index.update(oldValues[columnIndex], newValues[columnIndex], rowId);
        }
    }

    private SecondaryIndex findIndex(int tableId, String columnName) {
        List<SecondaryIndex> indexes = secondaryIndexes.get(tableId);
        if (indexes == null) {
            return null;
        }
        for (SecondaryIndex index : indexes) {
            if (index.getColumnName().equalsIgnoreCase(columnName)) {
                return index;
            }
        }
        return null;
    }

    @Override
    public void close() {
        // 目前无需额外操作
    }

    private static final class SecondaryIndex {
        private final IndexDefinition definition;
        private final ColumnDefinition column;
        private final Map<Object, List<Long>> buckets = new HashMap<>();

        private SecondaryIndex(IndexDefinition definition, ColumnDefinition column) {
            this.definition = definition;
            this.column = column;
        }

        int getColumnIndex() {
            return definition.getColumnIndex();
        }

        String getColumnName() {
            return column.getName();
        }

        void add(Object value, long rowId) {
            Object key = normalize(value);
            if (key == null) {
                return;
            }
            buckets.computeIfAbsent(key, k -> new ArrayList<>()).add(rowId);
        }

        void remove(Object value, long rowId) {
            Object key = normalize(value);
            if (key == null) {
                return;
            }
            List<Long> list = buckets.get(key);
            if (list != null) {
                list.remove(rowId);
                if (list.isEmpty()) {
                    buckets.remove(key);
                }
            }
        }

        void update(Object oldValue, Object newValue, long rowId) {
            Object oldKey = normalize(oldValue);
            Object newKey = normalize(newValue);
            if (Objects.equals(oldKey, newKey)) {
                return;
            }
            if (oldKey != null) {
                List<Long> list = buckets.get(oldKey);
                if (list != null) {
                    list.remove(rowId);
                    if (list.isEmpty()) {
                        buckets.remove(oldKey);
                    }
                }
            }
            if (newKey != null) {
                buckets.computeIfAbsent(newKey, k -> new ArrayList<>()).add(rowId);
            }
        }

        List<Long> lookup(Object value) {
            Object key = normalize(value);
            if (key == null) {
                return List.of();
            }
            List<Long> list = buckets.get(key);
            return list == null ? List.of() : new ArrayList<>(list);
        }

        private Object normalize(Object value) {
            if (value == null) {
                return null;
            }
            switch (column.getType()) {
                case INTEGER:
                    if (value instanceof Number) {
                        return ((Number) value).longValue();
                    }
                    return Long.parseLong(value.toString());
                case REAL:
                    if (value instanceof Number) {
                        return ((Number) value).doubleValue();
                    }
                    return Double.parseDouble(value.toString());
                case TEXT:
                default:
                    return value.toString();
            }
        }
    }

    public static class Filter {
        private final String column;
        private final Object expected;

        public Filter(String column, Object expected) {
            this.column = column;
            this.expected = expected;
        }

        public boolean matches(Row row) {
            int idx = row.getTable().findColumnIndex(column);
            if (idx < 0) {
                return false;
            }
            Object actual = row.getValues()[idx];
            return Objects.equals(expected, actual);
        }

        public String getColumn() {
            return column;
        }

        public Object getExpected() {
            return expected;
        }
    }

    private record RowMutation(long rowId, Object[] oldValues, Object[] newValues, byte[] payload) { }
}
